package com.atguigu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import com.atguigu.util.DruidDSUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;
import java.util.stream.Collectors;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private DruidDataSource druidDataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidDSUtil.getDruidDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        DruidPooledConnection connection = druidDataSource.getConnection();
        String sink_table = value.getString( "sink_table" );
        JSONObject data = value.getJSONObject( "data" );
        String sql = genUpsql( sink_table, data );
        System.out.println( "写入Phoenix的sql语句打印测试：" + sql );

        PreparedStatement preparedStatement = connection.prepareStatement( sql );
        preparedStatement.execute();
        connection.commit();
        connection.close();
    }

    // upsert into table key() values();
    private String genUpsql(String sinkTable, JSONObject data) {
        //获取列名
        Set<String> columns = data.keySet();
        //获取列值
        Collection<Object> values = data.values();
//        Collection collection = m1(values);
        return "upsert into " + GmallConfig.PHOENIX_DB + "." + sinkTable + "(" +
                StringUtils.join(columns, ",") + ") values('" +
                StringUtils.join(values, "','") + "')";
    }
//    public static Collection m1(Collection<Object> list){
//        return list.stream().map(o -> o.toString().replaceAll("'", "\\'")).collect( Collectors.toList());
//    }
}
